package org.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.atguigu.gmall.realtime.app.BaseApp;
import org.atguigu.gmall.realtime.bean.TradeProvinceOrderBean;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.function.AsyncDimFunction;
import org.atguigu.gmall.realtime.util.AtguiguUtil;
import org.atguigu.gmall.realtime.util.FlinkSinkUtil;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dws
 * Author: ZengHaiFeng
 * CreateTime:2023/5/6 11:48:45
 * Description: Todo
 * Version：1.0
 */
public class Dws_10_DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_10_DwsTradeProvinceOrderWindow()
                .init(40010,2,
                        Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,
                        "Dws_10_DwsTradeProvinceOrderWindow");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        // group by order id
        SingleOutputStreamOperator<TradeProvinceOrderBean> orderIdStream = groupByOrderId(ds);
        //orderIdStream.print();
        // group by order detail id.
        SingleOutputStreamOperator<TradeProvinceOrderBean> orderDetailIdStream = groupByOrderDetailId(orderIdStream);
        // window and aggregation.
        SingleOutputStreamOperator<TradeProvinceOrderBean> windowStream = windowAndAgg(orderDetailIdStream);
        //windowStream.print();
        // supplementation dimension.
        SingleOutputStreamOperator<TradeProvinceOrderBean> resultStream = streamWithDims(windowStream);
        //resultStream.print();
        // output to Clickhouse.
        writeToClickhouse(resultStream);

    }

    private void writeToClickhouse(SingleOutputStreamOperator<TradeProvinceOrderBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink(
                "dws_trade_province_order_window",
                TradeProvinceOrderBean.class
        ));
    }

    private SingleOutputStreamOperator<TradeProvinceOrderBean> streamWithDims(SingleOutputStreamOperator<TradeProvinceOrderBean> windowStream) {
        return AsyncDataStream.unorderedWait(windowStream,
                new AsyncDimFunction<TradeProvinceOrderBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_province";
                    }

                    @Override
                    public String getId(TradeProvinceOrderBean bean) {
                        return bean.getProvinceId();
                    }

                    @Override
                    public void addDim(TradeProvinceOrderBean bean, JSONObject dim) {
                        bean.setProvinceName(dim.getString("NAME"));
                    }
                }, 60, TimeUnit.SECONDS );
    }

    private SingleOutputStreamOperator<TradeProvinceOrderBean> windowAndAgg(SingleOutputStreamOperator<TradeProvinceOrderBean> orderDetailIdStream) {
        return orderDetailIdStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean,ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(60))
        )
                .keyBy(TradeProvinceOrderBean::getProvinceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(
                            TradeProvinceOrderBean value1,
                            TradeProvinceOrderBean value2) throws Exception {
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                }, new ProcessWindowFunction<
                        TradeProvinceOrderBean,
                        TradeProvinceOrderBean,
                        String,
                        TimeWindow>() {
                    @Override
                    public void process(
                            String s,
                            Context context,
                            Iterable<TradeProvinceOrderBean> elements,
                            Collector<TradeProvinceOrderBean> out) throws Exception {
                        TradeProvinceOrderBean bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(context.window().getEnd()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                });
    }

    private SingleOutputStreamOperator<TradeProvinceOrderBean> groupByOrderDetailId(SingleOutputStreamOperator<TradeProvinceOrderBean> orderIdStream) {
        return orderIdStream.keyBy(TradeProvinceOrderBean::getOrderDetailId)
                .process(new KeyedProcessFunction<
                        String,
                        TradeProvinceOrderBean,
                        TradeProvinceOrderBean>() {
                    private ValueState<Boolean> isFirstOrderDetailState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<Boolean> isFirstOrderDetailDesc = new ValueStateDescriptor<>("isFirstOrderDetail", Boolean.class);
                        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(
                                org.apache.flink.api.common.time.Time.seconds(10))
                                .updateTtlOnReadAndWrite()
                                .useProcessingTime()
                                .build();
                        isFirstOrderDetailDesc.enableTimeToLive(stateTtlConfig);
                        isFirstOrderDetailState = getRuntimeContext().getState(isFirstOrderDetailDesc);
                    }

                    @Override
                    public void processElement(
                            TradeProvinceOrderBean value,
                            Context ctx,
                            Collector<TradeProvinceOrderBean> out) throws Exception {
                        Boolean isFirstOrderDetail = isFirstOrderDetailState.value();
                        if (isFirstOrderDetail == null){
                            isFirstOrderDetailState.update(false);
                            out.collect(value);
                        }
                    }
                });
    }

    private SingleOutputStreamOperator<TradeProvinceOrderBean> groupByOrderId(DataStreamSource<String> ds) {
        return ds.map(JSON::parseObject)
                .keyBy(obj -> obj.getString("order_id"))
                .process(new KeyedProcessFunction<String, JSONObject, TradeProvinceOrderBean>() {
                    private ValueState<Boolean> isFirstOrderState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<Boolean> isFirstOrderDesc = new ValueStateDescriptor<>("isFirstOrder",
                                Boolean.class);
                        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(
                                org.apache.flink.api.common.time.Time.seconds(10))
                                .updateTtlOnReadAndWrite()
                                .useProcessingTime()
                                .build();
                        isFirstOrderDesc.enableTimeToLive(stateTtlConfig);
                        isFirstOrderState = getRuntimeContext().getState(isFirstOrderDesc);
                    }

                    @Override
                    public void processElement(
                            JSONObject value,
                            Context ctx,
                            Collector<TradeProvinceOrderBean> out) throws Exception {
                        Boolean isFirstOrder = isFirstOrderState.value();
                        Long orderCount = 0L;
                        if (isFirstOrder == null){
                            orderCount = 1L;
                            isFirstOrderState.update(false);
                        }
                        out.collect(TradeProvinceOrderBean.builder()
                                        .orderDetailId(value.getString("id"))
                                        .provinceId(value.getString("province_id"))
                                        .orderCount(orderCount)
                                        .orderAmount(value.getBigDecimal("split_total_amount"))
                                        .ts(value.getLong("ts") * 1000)
                                .build());
                    }
                });
    }
}
